{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "consistent-evidence",
   "metadata": {},
   "outputs": [],
   "source": [
    "from flink.plan.Environment import get_environment\n",
    "from flink.functions.GroupReduceFunction import GroupReduceFunction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "strong-slovenia",
   "metadata": {},
   "outputs": [],
   "source": [
    "#pip install apache-flink==1.10.1\n",
    "#pip install flink"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "deluxe-hospital",
   "metadata": {},
   "outputs": [],
   "source": [
    "# pip install --upgrade pip"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "welcome-satin",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Adder(GroupReduceFunction):\n",
    "    def reduce(self, iterator, collector):\n",
    "        count, word = iterator.next()\n",
    "        count += sum([x[0] for x in iterator])\n",
    "        collector.collect((count, word))\n",
    "\n",
    "\n",
    "# 1. \n",
    "        env = get_environment() \n",
    "\n",
    "# 2. \n",
    "        data = env.from_elements(\"Who's there?\",\n",
    "        \"I think I hear them. Stand, ho! Who's there?\")\n",
    "            \n",
    "# 3. \n",
    "        data \\\n",
    "          .flat_map(lambda x, c: [(1, word) for word in x.lower().split()]) \\\n",
    "          .group_by(1) \\\n",
    "          .reduce_group(Adder(), combinable=True) \\\n",
    "          .output()\n",
    "        \n",
    "# 4. \n",
    "        env.execute(local=True) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "boring-magazine",
   "metadata": {},
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "Invalid mode specified: ",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-11-0247cb629583>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     14\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     15\u001b[0m \u001b[0;31m# 4.\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 16\u001b[0;31m \u001b[0menv\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mlocal\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;32mTrue\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m~/.pyenv/versions/3.7.0/lib/python3.7/site-packages/flink/plan/Environment.py\u001b[0m in \u001b[0;36mexecute\u001b[0;34m(self, local)\u001b[0m\n\u001b[1;32m    214\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_optimize_plan\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    215\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 216\u001b[0;31m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_container\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mis_planning\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    217\u001b[0m             \u001b[0mport\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msys\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstdin\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mreadline\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrstrip\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'\\n'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    218\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_connection\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mConnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPureTCPConnection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mport\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.7.0/lib/python3.7/site-packages/flink/plan/Environment.py\u001b[0m in \u001b[0;36mis_planning\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     55\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_plan_mode\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mFalse\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     56\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 57\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mValueError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Invalid mode specified: \"\u001b[0m \u001b[0;34m+\u001b[0m \u001b[0mmode\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     58\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_plan_mode\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     59\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mValueError\u001b[0m: Invalid mode specified: "
     ]
    }
   ],
   "source": [
    "\n",
    "        \n",
    "# 4. \n",
    "env.execute(local=True) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "respective-vinyl",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<flink.plan.DataSet.DataSink at 0x7fafe9b08828>"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\n",
    "# 3. \n",
    "data.flat_map(lambda x, c: [(1, word) for word in x.lower().split()]) \\\n",
    "  .group_by(1) \\\n",
    "  .reduce_group(Adder(), combinable=True) \\\n",
    "  .output()\n",
    "            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "aquatic-moscow",
   "metadata": {},
   "outputs": [
    {
     "ename": "ValueError",
     "evalue": "Invalid mode specified: ",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mValueError\u001b[0m                                Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-19-28a1397ae01b>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0menv\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexecute\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;32m~/.pyenv/versions/3.7.0/lib/python3.7/site-packages/flink/plan/Environment.py\u001b[0m in \u001b[0;36mexecute\u001b[0;34m(self, local)\u001b[0m\n\u001b[1;32m    214\u001b[0m         \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_optimize_plan\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    215\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m--> 216\u001b[0;31m         \u001b[0;32mif\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_container\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mis_planning\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m    217\u001b[0m             \u001b[0mport\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msys\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mstdin\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mreadline\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mrstrip\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'\\n'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m    218\u001b[0m             \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_connection\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mConnection\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mPureTCPConnection\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mport\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;32m~/.pyenv/versions/3.7.0/lib/python3.7/site-packages/flink/plan/Environment.py\u001b[0m in \u001b[0;36mis_planning\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m     55\u001b[0m                 \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_plan_mode\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0;32mFalse\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     56\u001b[0m             \u001b[0;32melse\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 57\u001b[0;31m                 \u001b[0;32mraise\u001b[0m \u001b[0mValueError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m\"Invalid mode specified: \"\u001b[0m \u001b[0;34m+\u001b[0m \u001b[0mmode\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m     58\u001b[0m         \u001b[0;32mreturn\u001b[0m \u001b[0mself\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0m_plan_mode\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m     59\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mValueError\u001b[0m: Invalid mode specified: "
     ]
    }
   ],
   "source": [
    "env.execute() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "radical-grove",
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'pyflink.table.expressions'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-17-fbdff9387a6c>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mDataTypes\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mTableEnvironment\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mEnvironmentSettings\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      2\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mdescriptors\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mSchema\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mOldCsv\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mFileSystem\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 3\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexpressions\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mlit\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      4\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0msettings\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mEnvironmentSettings\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mnew_instance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0min_batch_mode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0muse_blink_planner\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mbuild\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mModuleNotFoundError\u001b[0m: No module named 'pyflink.table.expressions'"
     ]
    }
   ],
   "source": [
    "from pyflink.table import DataTypes, TableEnvironment, EnvironmentSettings\n",
    "from pyflink.table.descriptors import Schema, OldCsv, FileSystem\n",
    "from pyflink.table.expressions import lit\n",
    "\n",
    "settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()\n",
    "t_env = TableEnvironment.create(settings)\n",
    "\n",
    "# write all the data to one file\n",
    "t_env.get_config().get_configuration().set_string(\"parallelism.default\", \"1\")\n",
    "t_env.connect(FileSystem().path('/tmp/input')) \\\n",
    "    .with_format(OldCsv()\n",
    "                 .field('word', DataTypes.STRING())) \\\n",
    "    .with_schema(Schema()\n",
    "                 .field('word', DataTypes.STRING())) \\\n",
    "    .create_temporary_table('mySource')\n",
    "\n",
    "t_env.connect(FileSystem().path('/tmp/output')) \\\n",
    "    .with_format(OldCsv()\n",
    "                 .field_delimiter('\\t')\n",
    "                 .field('word', DataTypes.STRING())\n",
    "                 .field('count', DataTypes.BIGINT())) \\\n",
    "    .with_schema(Schema()\n",
    "                 .field('word', DataTypes.STRING())\n",
    "                 .field('count', DataTypes.BIGINT())) \\\n",
    "    .create_temporary_table('mySink')\n",
    "\n",
    "tab = t_env.from_path('mySource')\n",
    "tab.group_by(tab.word) \\\n",
    "   .select(tab.word, lit(1).count) \\\n",
    "   .execute_insert('mySink').wait()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "similar-camel",
   "metadata": {},
   "outputs": [
    {
     "ename": "SyntaxError",
     "evalue": "invalid syntax (<ipython-input-18-f7adcbf76e92>, line 1)",
     "output_type": "error",
     "traceback": [
      "\u001b[0;36m  File \u001b[0;32m\"<ipython-input-18-f7adcbf76e92>\"\u001b[0;36m, line \u001b[0;32m1\u001b[0m\n\u001b[0;31m    EnvironmentSettings settings = EnvironmentSettings.newInstance().build();\u001b[0m\n\u001b[0m                               ^\u001b[0m\n\u001b[0;31mSyntaxError\u001b[0m\u001b[0;31m:\u001b[0m invalid syntax\n"
     ]
    }
   ],
   "source": [
    "EnvironmentSettings settings = EnvironmentSettings.newInstance().build();\n",
    "TableEnvironment tEnv = TableEnvironment.create(settings);\n",
    "\n",
    "tEnv.executeSql(\"CREATE TABLE transactions (\\n\" +\n",
    "    \"    account_id  BIGINT,\\n\" +\n",
    "    \"    amount      BIGINT,\\n\" +\n",
    "    \"    transaction_time TIMESTAMP(3),\\n\" +\n",
    "    \"    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND\\n\" +\n",
    "    \") WITH (\\n\" +\n",
    "    \"    'connector' = 'kafka',\\n\" +\n",
    "    \"    'topic'     = 'transactions',\\n\" +\n",
    "    \"    'properties.bootstrap.servers' = 'kafka:9092',\\n\" +\n",
    "    \"    'format'    = 'csv'\\n\" +\n",
    "    \")\");\n",
    "\n",
    "tEnv.executeSql(\"CREATE TABLE spend_report (\\n\" +\n",
    "    \"    account_id BIGINT,\\n\" +\n",
    "    \"    log_ts     TIMESTAMP(3),\\n\" +\n",
    "    \"    amount     BIGINT\\n,\" +\n",
    "    \"    PRIMARY KEY (account_id, log_ts) NOT ENFORCED\" +\n",
    "    \") WITH (\\n\" +\n",
    "    \"   'connector'  = 'jdbc',\\n\" +\n",
    "    \"   'url'        = 'jdbc:mysql://mysql:3306/sql-demo',\\n\" +\n",
    "    \"   'table-name' = 'spend_report',\\n\" +\n",
    "    \"   'driver'     = 'com.mysql.jdbc.Driver',\\n\" +\n",
    "    \"   'username'   = 'sql-demo',\\n\" +\n",
    "    \"   'password'   = 'demo-sql'\\n\" +\n",
    "    \")\");\n",
    "\n",
    "Table transactions = tEnv.from(\"transactions\");\n",
    "report(transactions).executeInsert(\"spend_report\");"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "egyptian-brunei",
   "metadata": {},
   "outputs": [],
   "source": [
    "private static class Event {\n",
    "    public final String key;\n",
    "    public final long timestamp;\n",
    "    ...\n",
    "}\n",
    "\n",
    "public static void main(String[] args) throws Exception {\n",
    "    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();\n",
    "  \n",
    "    env.addSource(new EventSource())\n",
    "        .keyBy(e -> e.key)\n",
    "        .flatMap(new Deduplicator())\n",
    "        .print();\n",
    "  \n",
    "    env.execute();\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
